package edu.utexas.ipc;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

public class IPCConnection implements Connection {

	String tag;
	private BlockingQueue<Message> queue;
	private Multiplexer multiplexer;
	private MessageListener listener;
	private Thread listenerThread;
	
	IPCConnection(String tag, BlockingQueue<Message> queue, Multiplexer multiplexer) {
		this.tag = tag;
		this.queue = queue;
		this.multiplexer = multiplexer;
		this.listenerThread = null;
	}
	
	public Message receive() throws InterruptedException {
		return queue.take();
	}

	public Message receive(int timeout) throws InterruptedException {
		return queue.poll(timeout, TimeUnit.MILLISECONDS);
	}
	
	public void send(String message) {
		multiplexer.broadcast(tag, message);
	}

	public void send(int destID, String message) {
		multiplexer.sendMessage(destID, tag, message);
	}
	
	public void close() {
		System.out.println("close connection: " + tag);
		if (listenerThread != null) {
			listenerThread.interrupt();
		}
		multiplexer.closeConnection(tag);
	}
	
	public void registerListener(final MessageListener msglistener) {
		
		this.listener = msglistener;
		this.listenerThread = new Thread() {
			public void run() {
				//System.out.println("start listener thread for " + tag);
				try {
					while(true) {
						Message msg = queue.take();
						if(msg.tag.equals("connection_lost")) {
							// System.out.println("[" + tag + "] connection_lost " + msg.sourceID);
							listener.connectionLost(msg.sourceID);
							continue;
						}
						listener.messageReceived(msg);
					}
				} catch (InterruptedException e) {
					//System.out.println("stop listener thread for " + tag);
				}
			}
		};
		listenerThread.setDaemon(true);
		listenerThread.start();
	}

	public boolean isConnected(int processID) {
		return multiplexer.isConnected(processID);
	}
}
